/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase;

import com.facebook.presto.spi.SchemaTableName;
import com.google.common.collect.ImmutableMap;
import io.airlift.log.Logger;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.util.Bytes;

import javax.inject.Inject;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * HBase tables
 *
 * @author wupeng
 * @date 2019/01/29
 */
public class HBaseTables {

    public static final Logger logger = Logger.get(HBaseTables.class);

    private final HBaseClientManager hbaseClientManager;

    @Inject
    public HBaseTables(HBaseClientManager hbaseClientManager) {
        this.hbaseClientManager = hbaseClientManager;
    }


    /**
     * 从 Hbase Meta 中获取所有表信息
     * @param schema
     * @return
     */
    Map<SchemaTableName, HBaseTableHandle> getTables(String schema) {
        if("information_schema".equalsIgnoreCase(schema)) {
            // presto 内部管理 schema
            return Collections.emptyMap();
        }
        Map<SchemaTableName, HBaseTableHandle> tables;
        try(Admin admin = hbaseClientManager.getAdmin()) {
            ImmutableMap.Builder<SchemaTableName, HBaseTableHandle> tablesBuilder = ImmutableMap.builder();
            List<TableDescriptor> descriptors = admin.listTableDescriptorsByNamespace(schema.getBytes(StandardCharsets.UTF_8));
            for (TableDescriptor table : descriptors) {
                // If the target table is in the other namespace, table.getNameAsString() will return
                // value like 'namespace1:tableName1', so we have to remove the unnecessary namespace.
                String tableName;
                final TableName tableTableName = table.getTableName();
                String originName = tableTableName.getNameAsString();
                if (originName != null && originName.contains(":")) {
                    tableName = originName.split(":")[1];
                } else {
                    tableName = originName;
                }

                Objects.requireNonNull(tableName, "tableName cannot be null!");
                SchemaTableName schemaTableName = new SchemaTableName(schema, tableName);
                tablesBuilder.put(schemaTableName, new HBaseTableHandle(schemaTableName));
            }
            tables = tablesBuilder.build();
            return tables;
        } catch (Exception ex) {
            logger.error(ex, ex.getMessage());
        }
        return Collections.emptyMap();
    }


    /**
     * 获取 HBase namespace
     * @return
     */
    Set<String> getSchemaNames() {
        NamespaceDescriptor[] namespaceDescriptors = new NamespaceDescriptor[0];
        try (Admin admin = hbaseClientManager.getAdmin();) {
            namespaceDescriptors = admin.listNamespaceDescriptors();
        } catch (IOException e) {
            logger.error(e, e.getMessage());
        }

        HashSet<String> set = new HashSet<>();
        NamespaceDescriptor[] temp = namespaceDescriptors;
        int namespaceDescriptorLength = namespaceDescriptors.length;

        for (int i = 0; i < namespaceDescriptorLength; ++i) {
            NamespaceDescriptor namespaceDescriptor = temp[i];
            set.add(namespaceDescriptor.getName());
        }

        return set;
    }

    public boolean dropTable(String schema, String tableName) {
        if("information_schema".equalsIgnoreCase(schema)) {
            // presto 内部管理 schema
            return false;
        }
        try (Admin admin = hbaseClientManager.getAdmin();) {
            admin.disableTable(TableName.valueOf(schema + ":" + tableName));
            admin.deleteTable(TableName.valueOf(schema + ":" + tableName));
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return false;
        }
        return true;
    }

}
